/*
 * Copyright 2017-2025 Lenses.io Ltd
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.lenses.streamreactor.connect.cassandra

import io.lenses.streamreactor.common.config.base.traits._
import org.apache.kafka.common.config.ConfigDef
import org.apache.kafka.common.config.ConfigDef.Importance
import org.apache.kafka.common.config.ConfigDef.Type

/**
 * Holds the base configuration.
 */
object CassandraConfig {

  val configDef: ConfigDef = new ConfigDef()
    .define(
      CassandraConfigConstants.CONTACT_POINTS,
      Type.STRING,
      CassandraConfigConstants.CONTACT_POINT_DEFAULT,
      Importance.HIGH,
      CassandraConfigConstants.CONTACT_POINT_DOC,
      "Connection",
      1,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.CONTACT_POINTS,
    )
    .define(
      CassandraConfigConstants.PORT,
      Type.INT,
      CassandraConfigConstants.PORT_DEFAULT,
      ConfigDef.Range.atLeast(1),
      Importance.HIGH,
      CassandraConfigConstants.PORT_DOC,
      "Connection",
      2,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.PORT,
    ).define(
      CassandraConfigConstants.MAX_CONCURRENT_REQUESTS,
      Type.INT,
      CassandraConfigConstants.MAX_CONCURRENT_REQUESTS_DEFAULT,
      ConfigDef.Range.atLeast(1),
      Importance.MEDIUM,
      CassandraConfigConstants.MAX_CONCURRENT_REQUESTS_DOC,
      "Connection",
      3,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.MAX_CONCURRENT_REQUESTS,
    )
    .define(
      CassandraConfigConstants.CONNECTION_POOL_SIZE,
      Type.INT,
      CassandraConfigConstants.CONNECTION_POOL_SIZE_DEFAULT,
      ConfigDef.Range.atLeast(1),
      Importance.MEDIUM,
      CassandraConfigConstants.CONNECTION_POOL_SIZE_DOC,
      "Connection",
      4,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.CONNECTION_POOL_SIZE,
    )
    .define(
      CassandraConfigConstants.COMPRESSION,
      Type.STRING,
      CassandraConfigConstants.COMPRESSION_DEFAULT,
      ConfigDef.CaseInsensitiveValidString.in("NONE", "SNAPPY", "SNAPPY"),
      Importance.MEDIUM,
      CassandraConfigConstants.COMPRESSION_DOC,
      "Connection",
      5,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.COMPRESSION,
    )
    .define(
      CassandraConfigConstants.QUERY_TIMEOUT,
      Type.INT,
      CassandraConfigConstants.QUERY_TIMEOUT_DEFAULT,
      ConfigDef.Range.atLeast(1),
      Importance.MEDIUM,
      CassandraConfigConstants.QUERY_TIMEOUT_DOC,
      "Connection",
      6,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.QUERY_TIMEOUT,
    )
    .define(
      CassandraConfigConstants.MAX_BATCH_SIZE,
      Type.INT,
      CassandraConfigConstants.MAX_BATCH_SIZE_DEFAULT,
      ConfigDef.Range.atLeast(1),
      Importance.MEDIUM,
      CassandraConfigConstants.MAX_BATCH_SIZE_DOC,
      "Connection",
      7,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.MAX_BATCH_SIZE,
    )
    .define(
      CassandraConfigConstants.LOAD_BALANCING_LOCAL_DC,
      Type.STRING,
      "",
      Importance.MEDIUM,
      CassandraConfigConstants.LOAD_BALANCING_LOCAL_DC_DOC,
      "Connection",
      8,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.LOAD_BALANCING_LOCAL_DC,
    )
    .define(
      CassandraConfigConstants.SSL_ENABLED,
      Type.BOOLEAN,
      CassandraConfigConstants.SSL_ENABLED_DEFAULT,
      Importance.LOW,
      CassandraConfigConstants.SSL_ENABLED_DOC,
      "Connection",
      9,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_ENABLED,
    )
    .define(
      CassandraConfigConstants.SSL_TRUST_STORE_PATH,
      Type.STRING,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_TRUST_STORE_PATH_DOC,
      "Connection",
      10,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_TRUST_STORE_PATH,
    )
    .define(
      CassandraConfigConstants.SSL_TRUST_STORE_PASSWD,
      Type.PASSWORD,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_TRUST_STORE_PASSWD_DOC,
      "Connection",
      11,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_TRUST_STORE_PASSWD,
    )
    .define(
      CassandraConfigConstants.SSL_KEY_STORE_PATH,
      Type.STRING,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_KEY_STORE_PATH_DOC,
      "Connection",
      12,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_KEY_STORE_PATH,
    )
    .define(
      CassandraConfigConstants.SSL_KEY_STORE_PASSWD,
      Type.PASSWORD,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_KEY_STORE_PASSWD_DOC,
      "Connection",
      13,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_KEY_STORE_PASSWD,
    )
    .define(
      CassandraConfigConstants.SSL_PROVIDER,
      Type.STRING,
      CassandraConfigConstants.SSL_PROVIDER_DEFAULT,
      ConfigDef.CaseInsensitiveValidString.in("None", "JDK", "OpenSSL"),
      Importance.MEDIUM,
      CassandraConfigConstants.SSL_PROVIDER_DOC,
      "Connection",
      14,
      ConfigDef.Width.MEDIUM,
      CassandraConfigConstants.SSL_PROVIDER,
    )
    .define(
      CassandraConfigConstants.SSL_CIPHER_SUITES,
      Type.STRING,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_CIPHER_SUITES_DOC,
      "Connection",
      15,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_CIPHER_SUITES,
    )
    .define(
      CassandraConfigConstants.SSL_HOSTNAME_VERIFICATION,
      Type.BOOLEAN,
      CassandraConfigConstants.SSL_HOSTNAME_VERIFICATION_DEFAULT,
      Importance.MEDIUM,
      CassandraConfigConstants.SSL_HOSTNAME_VERIFICATION_DOC,
      "Connection",
      16,
      ConfigDef.Width.MEDIUM,
      CassandraConfigConstants.SSL_HOSTNAME_VERIFICATION,
    )
    .define(
      CassandraConfigConstants.SSL_OPENSSL_KEY_CERT_CHAIN,
      Type.STRING,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_OPENSSL_KEY_CERT_CHAIN_DOC,
      "Connection",
      17,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_OPENSSL_KEY_CERT_CHAIN,
    )
    .define(
      CassandraConfigConstants.SSL_OPENSSL_PRIVATE_KEY,
      Type.STRING,
      "",
      Importance.LOW,
      CassandraConfigConstants.SSL_OPENSSL_PRIVATE_KEY_DOC,
      "Connection",
      18,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.SSL_OPENSSL_PRIVATE_KEY,
    )
    .define(
      CassandraConfigConstants.AUTH_PROVIDER,
      Type.STRING,
      CassandraConfigConstants.AUTH_PROVIDER_DEFAULT,
      ConfigDef.CaseInsensitiveValidString.in("None", "PLAIN", "GSSAPI"),
      Importance.HIGH,
      CassandraConfigConstants.AUTH_PROVIDER_DOC,
      "Authentication",
      1,
      ConfigDef.Width.MEDIUM,
      CassandraConfigConstants.AUTH_PROVIDER,
    )
    .define(
      CassandraConfigConstants.AUTH_USERNAME,
      Type.STRING,
      CassandraConfigConstants.AUTH_USERNAME_DEFAULT,
      Importance.HIGH,
      CassandraConfigConstants.AUTH_USERNAME_DOC,
      "Authentication",
      2,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.AUTH_USERNAME,
    )
    .define(
      CassandraConfigConstants.AUTH_PASSWORD,
      Type.PASSWORD,
      CassandraConfigConstants.AUTH_PASSWORD_DEFAULT,
      Importance.HIGH,
      CassandraConfigConstants.AUTH_PASSWORD_DOC,
      "Authentication",
      3,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.AUTH_PASSWORD,
    )
    .define(
      CassandraConfigConstants.AUTH_GSSAPI_KEYTAB,
      Type.STRING,
      CassandraConfigConstants.AUTH_GSSAPI_KEYTAB_DEFAULT,
      Importance.HIGH,
      CassandraConfigConstants.AUTH_GSSAPI_KEYTAB_DOC,
      "Authentication",
      4,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.AUTH_GSSAPI_KEYTAB,
    )
    .define(
      CassandraConfigConstants.AUTH_GSSAPI_PRINCIPAL,
      Type.STRING,
      CassandraConfigConstants.AUTH_GSSAPI_PRINCIPAL_DEFAULT,
      Importance.HIGH,
      CassandraConfigConstants.AUTH_GSSAPI_PRINCIPAL_DOC,
      "Authentication",
      5,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.AUTH_GSSAPI_PRINCIPAL,
    )
    .define(
      CassandraConfigConstants.AUTH_GSSAPI_SERVICE,
      Type.STRING,
      CassandraConfigConstants.AUTH_GSSAPI_SERVICE_DEFAULT,
      Importance.HIGH,
      CassandraConfigConstants.AUTH_GSSAPI_SERVICE_DOC,
      "Authentication",
      6,
      ConfigDef.Width.MEDIUM,
      CassandraConfigConstants.AUTH_GSSAPI_SERVICE,
    )
    .define(
      CassandraConfigConstants.IGNORE_ERRORS_MODE,
      Type.STRING,
      CassandraConfigConstants.IGNORE_ERRORS_MODE_DEFAULT,
      ConfigDef.CaseInsensitiveValidString.in("none", "all", "driver"),
      Importance.MEDIUM,
      CassandraConfigConstants.IGNORE_ERRORS_MODE_DOC,
      "Error",
      1,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.IGNORE_ERRORS_MODE,
    )
    .define(
      CassandraConfigConstants.NBR_OF_RETRIES,
      Type.INT,
      CassandraConfigConstants.NBR_OF_RETIRES_DEFAULT,
      Importance.MEDIUM,
      CassandraConfigConstants.NBR_OF_RETRIES_DOC,
      "Error",
      2,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.NBR_OF_RETRIES,
    )
    .define(
      CassandraConfigConstants.ERROR_RETRY_INTERVAL,
      Type.INT,
      CassandraConfigConstants.ERROR_RETRY_INTERVAL_DEFAULT,
      Importance.MEDIUM,
      CassandraConfigConstants.ERROR_RETRY_INTERVAL_DOC,
      "Error",
      3,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.ERROR_RETRY_INTERVAL,
    )
    .define(
      CassandraConfigConstants.KCQL,
      Type.STRING,
      Importance.HIGH,
      CassandraConfigConstants.KCQL_DOC,
      "Mappings",
      1,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.KCQL,
    )
    .define(
      CassandraConfigConstants.PROGRESS_COUNTER_ENABLED,
      Type.BOOLEAN,
      CassandraConfigConstants.PROGRESS_COUNTER_ENABLED_DEFAULT,
      Importance.MEDIUM,
      CassandraConfigConstants.PROGRESS_COUNTER_ENABLED_DOC,
      "Metrics",
      1,
      ConfigDef.Width.MEDIUM,
      CassandraConfigConstants.PROGRESS_COUNTER_ENABLED_DISPLAY,
    )
    .define(
      CassandraConfigConstants.ERROR_POLICY,
      Type.STRING,
      CassandraConfigConstants.ERROR_POLICY_DEFAULT,
      ConfigDef.CaseInsensitiveValidString.in("NOOP", "THROW", "RETRY"),
      Importance.HIGH,
      CassandraConfigConstants.ERROR_POLICY_DOC,
      "Error",
      4,
      ConfigDef.Width.LONG,
      CassandraConfigConstants.ERROR_POLICY,
    )

}

case class CassandraConfig(props: Map[String, String])
    extends BaseConfig(CassandraConfigConstants.CONNECTOR_PREFIX, CassandraConfig.configDef, props)
    with KcqlSettings
    with ErrorPolicySettings
    with NumberRetriesSettings
